package com.pk.flink.apps.operators;


import com.pk.flink.bean.Access;
import com.pk.flink.functions.source.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 使用Flink 实现单词计数
 * @author pk
 */
public class FlinkOperatorsApp {
    public static void main(String[] args) throws Exception {
//        //获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> accessLogStreamSource = env.readTextFile("data/access.log");

        //Map
        SingleOutputStreamOperator<Access> mapResults = accessLogStreamSource.map(new AccecssMapFunction());
        mapResults.print();

        //Filter
        SingleOutputStreamOperator<Access> filterResults1 = accessLogStreamSource.filter(new AccecssFilterFunction()).map(new AccecssMapFunction());
        filterResults1.print();

        SingleOutputStreamOperator<Access> filterResults2 = accessLogStreamSource.map(new AccecssMapFunction()).filter(x->x.getTraffic()>100);
        filterResults2.print();

        //FlatMap
        SingleOutputStreamOperator<Access> flatMapResults = accessLogStreamSource.flatMap(new AccecssFlatMapFunction());
        flatMapResults.print();

        //KeyBy
        SingleOutputStreamOperator<Access> trafficKeyBy = accessLogStreamSource.map(new AccecssMapFunction()).
                keyBy(new AccecssKeyByFunction()).sum("traffic");
        trafficKeyBy.print();


        //Reduce
        SingleOutputStreamOperator<Access> reduce = accessLogStreamSource.map(new AccecssMapFunction()).
                keyBy(new AccecssKeyByFunction()).reduce(new AccecssReduceFunction());
        reduce.print();

        DataStreamSource<Integer> stream1 = env.fromElements(1,2,3);
        DataStreamSource<Integer> stream2 = env.fromElements(11,12,13);
        DataStreamSource<String> stream3 = env.fromElements("A","B","C");
        //Union
        stream1.union(stream2).print();

        //Connect
        ConnectedStreams<Integer, String> connect = stream1.connect(stream3);
        connect.map(new MyCoMapFunction()).print();
        connect.flatMap(new MyCoFlatMapFunction()).print();


        //partition
        env.setParallelism(3);
        accessLogStreamSource.map(new AccecssMapFunction())
                .map(new MapFunction<Access, Tuple2<String,Access>>() {
                    @Override
                    public Tuple2<String, Access> map(Access access) throws Exception {
                        return Tuple2.of(access.getDomain(),access);
                    }
                })
                .partitionCustom(new AccessPartitionFunction(),x-> x.f0)
                .map(new MapFunction<Tuple2<String, Access>, Access>() {
                    @Override
                    public Access map(Tuple2<String, Access> tuple) throws Exception {
                        System.out.println("current Thread:" + Thread.currentThread().getId() + ",current value:" + tuple.f1.getDomain());
                        return tuple.f1;
                    }
                })
                .print();


        //side output
        SingleOutputStreamOperator<Access> mainStream = accessLogStreamSource.map(new AccecssMapFunction());
        OutputTag<Access> outputTagBeijing = new OutputTag<Access>("beijing"){};
        OutputTag<Access> outputTagTianJin = new OutputTag<Access>("tianjin"){};
        SingleOutputStreamOperator<Access> process = mainStream.process(new ProcessFunction<Access, Access>() {
            @Override
            public void processElement(Access access, Context context, Collector<Access> collector) throws Exception {
                if ("beijing".equalsIgnoreCase(access.getDomain())) {
                    context.output(outputTagBeijing, access);
                } else if ("tianjin".equalsIgnoreCase(access.getDomain())) {
                    context.output(outputTagTianJin, access);
                } else {
                    collector.collect(access);
                }
            }
        });
        process.print("Main Stram");
        process.getSideOutput(outputTagBeijing).print("BeiJing Stream");
        process.getSideOutput(outputTagTianJin).print("TianJin Stream");

        env.execute();
    }
}
